package redis

import (
	"context"
	"fmt"
	"github.com/gomodule/redigo/redis"
	"unsafe"
)

type (

	//订阅回调函数
	SubscribeCallback func(channel, massage string)

	Subscriber struct {
		cancel context.CancelFunc
		client *redis.PubSubConn
		cbMap  map[string]SubscribeCallback
	}
)

//连接
func (this *Subscriber) Connect(address string, password string) (err error) {
	var (
		conn redis.Conn
		ctx  context.Context
	)
	if conn, err = redis.Dial("tcp", address); err != nil {
		return
	}

	//验证redis密码
	if password != "" {
		if _, authErr := conn.Do("AUTH", password); authErr != nil {
			conn.Close()
			return fmt.Errorf("redis auth password error: %s", authErr)
		}
	}

	this.client = &redis.PubSubConn{Conn: conn}
	this.cbMap = make(map[string]SubscribeCallback)

	ctx, this.cancel = context.WithCancel(context.TODO())

	go this.Receive(ctx)

	return

}

func (this *Subscriber) Close() (err error) {
	if this.client == nil {
		return
	}
	if err = this.client.Close(); err != nil {
		return
	}

	//退出协程
	this.cancel()
	return
}

func (this *Subscriber) Subscribe(channel interface{}, cb SubscribeCallback) (err error) {
	if err = this.client.Subscribe(channel); err != nil {
		return
	}

	this.cbMap[channel.(string)] = cb

	return
}

func (this *Subscriber) Receive(ctx context.Context) {
	for {
		select {
		case <-ctx.Done():
			return
		default:
			switch res := this.client.Receive().(type) {
			case redis.Message:
				channel := (*string)(unsafe.Pointer(&res.Channel))
				message := (*string)(unsafe.Pointer(&res.Data))
				this.cbMap[*channel](*channel, *message)
			case redis.Subscription:
				//fmt.Printf("%s: %s %d\n", res.Channel, res.Kind, res.Count)
			case error:
				//log.Println("error handle...",res.Error())
				//time.Sleep(10*time.Second)
				continue
			}

		}

	}
}
